package io.openmessaging.storage;

import io.openmessaging.Conf;
import io.openmessaging.Message;
import io.openmessaging.Topic;
import io.openmessaging.io.FileChannelWrapper;
import io.openmessaging.map.TopicQueueOffsetMap;
import io.openmessaging.utils.Log;
import io.openmessaging.utils.Statistic;

import java.nio.ByteBuffer;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author cuijd    2021/9/26 9:11
 */
public class SSDStorage3 implements Storage {
    FileChannelWrapper[] channels = new FileChannelWrapper[4];
    AtomicLong[] waitSize = new AtomicLong[channels.length];
    AtomicLong[] maxWaitSize = new AtomicLong[channels.length];
    private TopicQueueOffsetMap<FileIndex> indices = new TopicQueueOffsetMap<>();

    public SSDStorage3() {
        for (int i = 0; i < channels.length; i++) {
            channels[i] = new FileChannelWrapper(Conf.ssdPath + i);
            waitSize[i] = new AtomicLong();
            maxWaitSize[i] = new AtomicLong(40 / channels.length);
            //根据已有数据恢复索引
            long readPos = 0;
            ByteBuffer head = ByteBuffer.allocate(9);//消息头，储存 topicId(1) queueId(2) offset(4) dataLength(2) 共9字节
            while (readPos < channels[i].size()) {
                head.clear();
                readPos += channels[i].readAndFlip(head, readPos);
                FileIndex index = indices.set(head.get(), head.getShort(), head.getInt(), new FileIndex((byte) i, readPos - 9, head.getShort()));
                readPos += index.length;
            }
        }

    }


    @Override
    public void save(Message m) {
        int fileId = m.topic.id % channels.length;
        ByteBuffer head = ByteBuffer.allocate(9);//消息头，储存 topicId(1) queueId(2) offset(4) dataLength(2) 共9字节
        head.clear();
        head.put(m.topic.id).putShort((short) m.queueId).putInt((int) m.offset).putShort((short) m.data.remaining()).flip();
        //写入ssd
        int dataSize = m.data.remaining();
        long position = channels[fileId].write(head, m.data);
        //消息索引
        FileIndex index = new FileIndex((byte) fileId, position, (short) dataSize);
        indices.set(m.topic.id, m.queueId, m.offset, index);

        long start = System.currentTimeMillis();
        synchronized (channels[fileId]) {
            long waits = waitSize[fileId].addAndGet(1);
            if (waits < maxWaitSize[fileId].get()) {
                try {
                    channels[fileId].wait(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            } else {
                channels[fileId].force();
                waitSize[fileId].set(0);
                channels[fileId].notifyAll();
            }
        }
        //最后几条消息如果不够 writeCount 条不会被唤醒，超时自动醒
        if (System.currentTimeMillis() - start > 900) {
            channels[fileId].force();
            Log.i("等待超时 ", m, " fileId ", fileId);
            maxWaitSize[fileId].getAndAdd(-1);
        }
    }

    @Override
    public ByteBuffer get(Topic topic, int queueId, long offset) {
        return get(topic, queueId, offset, null);
    }

    /**
     * 先判断有没有落盘，如果没有，返回null，否则先从缓存中取，如果缓存取不到，再从 ssd 读
     */
    public ByteBuffer get(Topic topic, int queueId, long offset, Storage cache) {
        FileIndex index = this.indices.get(topic.id, queueId, offset);
        if (index == null)
            return null;

        ByteBuffer readBuffer = null;
        if (cache != null)
            readBuffer = cache.get(topic, queueId, offset);

        if (readBuffer == null) {
            readBuffer = ByteBuffer.allocate(index.length);
            channels[index.fileId].readAndFlip(readBuffer, index.position + 9);//跳过header

            Statistic.SSD.readCount.add(1);
            Statistic.SSD.readSize.add(readBuffer.remaining());
        }
        return readBuffer;
    }

}
